/**
 * 
 */
package com.barisergun.memcached;

import static com.barisergun.memcached.TestVariables.*;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.testng.annotations.*;
import org.testng.*;

import net.spy.memcached.AddrUtil;
import net.spy.memcached.BinaryConnectionFactory;
import net.spy.memcached.MemcachedClient;

/**
 * @author barise
 * @date Dec 20, 2010
 * 
 */
public class MemcachedDistributedTests {

	private MemcachedClient memcacheClient;

	private int numberOfGivenServers;

	@BeforeClass
	public void setup() throws IOException {
		
		processSystemProperties();
		
		
		String addressesConcatenated = getDistributedServers();

		StringTokenizer tokenizer = new StringTokenizer(addressesConcatenated,
				" ");

		numberOfGivenServers = tokenizer.countTokens();

		memcacheClient = new MemcachedClient(new BinaryConnectionFactory(),
				AddrUtil.getAddresses(addressesConcatenated));
		
		

		

	}

	@AfterClass
	public void tearDown() {

		memcacheClient = null;
	}

	@Test
	public void testWithManyCacheEntriesInAsynchronizedStyle()
			throws InterruptedException, ExecutionException, TimeoutException {

		Long startTime = System.currentTimeMillis();

		DataObject[] locData = new DataObject[getEntryCount()];

		Map<String, Future<Boolean>> mapOfThreadFutures = new HashMap<String, Future<Boolean>>();

		for (Integer fakeMsIsdn = 0; fakeMsIsdn < getEntryCount(); fakeMsIsdn++) {

			locData[fakeMsIsdn] = new DataObject();

			Double xCoord = fakeMsIsdn / 10.00D;
			Double yCoord = fakeMsIsdn / 20.00D;
			locData[fakeMsIsdn].setXValue(xCoord);
			locData[fakeMsIsdn].setZValue(yCoord);
			Date dateNow = new Date();
			locData[fakeMsIsdn].setTimeStamp(dateNow);
			mapOfThreadFutures.put(fakeMsIsdn.toString(), memcacheClient.set(
					fakeMsIsdn.toString(), getTtlInSeconds(),
					locData[fakeMsIsdn]));

		}

		Long endTime = System.currentTimeMillis();

		Long executionTime = endTime - startTime;

		double tps = (getEntryCount() * 1000) / executionTime;

		System.out.println(String.format(DISTRIBUTED_REPORT_INFO_STRING,
				MEMCACHED_ASYNCH_TEST_INFO_STRING, INSERT_STRING,
				getEntryCount(), executionTime, numberOfGivenServers, tps));

		startTime = System.currentTimeMillis();

		for (Entry<String, Future<Boolean>> futureEntry : mapOfThreadFutures
				.entrySet()) {

			Assert.assertTrue(futureEntry.getValue().get(getGetTimeout(), TimeUnit.SECONDS),
					String
							.format(FAILED_TO_INSERT_STRING, futureEntry
									.getKey()));

		}

		endTime = System.currentTimeMillis();

		executionTime = endTime - startTime;

		System.out.println(String.format(
				REPORT_CONCURRENT_CACHES_TERMINATED_STRING, getEntryCount(),
				executionTime));

		startTime = System.currentTimeMillis();

		for (Integer entryId = 0; entryId < getEntryCount(); entryId++) {

			DataObject gotLocDataBack = (DataObject) memcacheClient
					.get(entryId.toString());

			if (gotLocDataBack == null) {

				throw new NullPointerException("Exception in index "
						+ entryId.toString());
			}

			Assert.assertEquals(gotLocDataBack.getXValue(), locData[entryId]
					.getXValue());
			Assert.assertEquals(gotLocDataBack.getZValue(), locData[entryId]
					.getZValue());
			Assert.assertEquals(gotLocDataBack.getTimeStamp(), locData[entryId]
					.getTimeStamp());

		}

		endTime = System.currentTimeMillis();

		executionTime = endTime - startTime;

		tps = (getEntryCount() * 1000) / executionTime;

		System.out.println(String.format(DISTRIBUTED_REPORT_INFO_STRING,
				MEMCACHED_ASYNCH_TEST_INFO_STRING, READING_STRING,
				getEntryCount(), executionTime, numberOfGivenServers, tps));
	}

}
